"""
Case Type   : Logical_Replication
Case Name   : 三权分立时,安全管理员和审计管理员用户创建、开启和删除逻辑复制槽失败，
    `       合理报错
Create At   : 2024/1/5
Owner       : xiaqi14
Description :
    1.修改参数
    2.创建sysadmin用户，配置用户白名单
    3.创建逻辑复制槽
    4.开启逻辑复制槽
    5.删除逻辑复制槽
    6.删除用户
    7.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.创建用户成功，配置白名单成功
    3.创建失败，合理报错
    4.开启失败，合理报错
    5.删除失败，合理报错
    6.删除成功
    7.恢复成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant


class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.sh_primary = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.user_name = 'u_logical_replication_0065'
        self.user_privileges = {f'{self.user_name}_audit': 'auditadmin',
                                f'{self.user_name}_create': 'createrole'}
        self.slot_name = 'slot_logical_replication_0065'
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)
        self.config_duty = 'enableSeparationOfDuty'
        self.duty_default = self.common.show_param(self.config_duty)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        if 'on' != self.duty_default:
            result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_duty}=on')
            self.assertTrue(result, '执行失败' + text)
            restart_res = self.sh_primary.restart_db_cluster()
            self.assertTrue(restart_res, '执行失败' + text)
        show_msg = self.common.show_param(self.config_duty)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        text = "-----step2:创建sysadmin用户，配置用户白名单，" \
               "expect:创建用户成功，配置白名单成功-----"
        self.logger.info(text)
        for user, privilege in self.user_privileges.items():
            sql_cmd = f'''drop user if exists {user};
                create user {user} {privilege} 
                password '{macro.COMMON_PASSWD}';'''
            sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
            self.logger.info(sql_msg)
            self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG and
                          self.constant.CREATE_ROLE_SUCCESS_MSG,
                          sql_msg, '执行失败:' + text)
            param_1 = f'local replication {user}              sha256'
            param_2 = f'host  replication {user} 127.0.0.1/32 sha256'
            param_3 = f'host  replication {user} ::1/128      sha256'
            result_1 = self.sh_primary.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                '', pghba_param=param_1)
            result_2 = self.sh_primary.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                '', pghba_param=param_2)
            result_3 = self.sh_primary.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                '', pghba_param=param_3)
            self.assertTrue(result_1 and result_2 and result_3,
                            '执行失败' + text)

        text = "-----step3:创建逻辑复制槽;expect:创建失败，合理报错-----"
        self.logger.info(text)
        for user in self.user_privileges.keys():
            create_slot_msg = self.sh_primary.exec_pg_recvlogical(
                '--create', 'postgres', self.slot_name, user, '-s 2 -W')
            self.logger.info(create_slot_msg)
            self.assertIn("FATAL:  must be replication role or a member of "
                          "the gs_role_replication role to use replication "
                          "slots when separation of privileges is used",
                          create_slot_msg, '执行失败:' + text)

        text = "-----step4:开启逻辑复制槽;expect:开启失败，合理报错-----"
        self.logger.info(text)
        for user in self.user_privileges.keys():
            start_slot_msg = self.sh_primary.exec_pg_recvlogical(
                '--start', 'postgres', self.slot_name, user,
                '-f - -o skip-empty-xacts=true -F 1 -s 2 -W')
            self.logger.info(start_slot_msg)
            self.assertIn("FATAL:  must be replication role or a member of "
                          "the gs_role_replication role to use replication "
                          "slots when separation of privileges is used",
                          start_slot_msg, '执行失败:' + text)

        text = "-----step5:删除逻辑复制槽;expect:删除失败，合理报错-----"
        self.logger.info(text)
        for user in self.user_privileges.keys():
            drop_slot_msg = self.sh_primary.exec_pg_recvlogical(
                '--drop', 'postgres', self.slot_name, user, '-s 2 -W')
            self.logger.info(drop_slot_msg)
            self.assertIn("FATAL:  must be replication role or a member of "
                          "the gs_role_replication role to use replication "
                          "slots when separation of privileges is used",
                          drop_slot_msg, '执行失败:' + text)

    def tearDown(self):
        text_1 = "-----step6:删除用户;expect:删除成功-----"
        self.logger.info(text_1)
        drop_msg = []
        for user in self.user_privileges.keys():
            drop_cmd = f"drop user {user} cascade; "
            msg = self.sh_primary.execut_db_sql(drop_cmd, dbname='postgres')
            self.logger.info(msg)
            drop_msg.append(msg)
        self.logger.info(drop_msg)

        text_2 = "-----step7:恢复参数，恢复pg_hba.conf文件，expect:恢复成功-----"
        self.logger.info(text_2)
        guc_result = []
        for user in self.user_privileges.keys():
            param_1 = f'local replication {user} '
            param_2 = f'host  replication {user} 127.0.0.1/32 '
            param_3 = f'host  replication {user} ::1/128 '
            result_1 = self.sh_primary.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                '', pghba_param=param_1)
            guc_result.append(result_1)
            result_2 = self.sh_primary.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                '', pghba_param=param_2)
            guc_result.append(result_2)
            result_3 = self.sh_primary.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                '', pghba_param=param_3)
            guc_result.append(result_3)
        self.logger.info(guc_result)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        current_duty = self.common.show_param(self.config_duty)
        if self.duty_default != current_duty:
            self.guc_duty_result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_duty}={self.duty_default}')
            self.restart_res = self.sh_primary.restart_db_cluster()
        show_duty_msg = self.common.show_param(self.config_duty)

        self.assertEqual(2, drop_msg.count(self.constant.DROP_ROLE_SUCCESS_MSG),
                         '执行失败:' + text_1)
        guc_expect = [True, True, True, True, True, True]
        self.assertEquals(guc_expect, guc_result, '执行失败' + text_2)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_2)
        self.assertEquals(show_duty_msg, self.duty_default, '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
